package com.vip.kafka.service.impl;

import com.vip.kafka.config.MessageProducer;
import com.vip.kafka.response.ErrorCode;
import com.vip.kafka.response.Result;
import com.vip.kafka.response.Results;
import com.vip.kafka.service.IKafkaService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.Set;

/**
 * Kafka Service 实现类
 *
 * @author wgb
 * @date 2020/5/19 15:40
 */
@Slf4j
@Service
public class KafkaServiceImpl implements IKafkaService {
    @Autowired
    private MessageProducer messageProducer;

    @Override
    public Result<String> createTopic(String topic) {
        try {
            Properties prop = new Properties();
            prop.put("bootstrap.servers", "192.168.25.93:9092");
            AdminClient admin = AdminClient.create(prop);
            ArrayList<NewTopic> topics = new ArrayList<>();
            // 创建主题  参数：主题名称、分区数、副本数
            NewTopic newTopic = new NewTopic(topic, 1, (short) 1);
            topics.add(newTopic);
            CreateTopicsResult result = admin.createTopics(topics);
            result.all().get();
            return Results.newSuccessResult(null, "创建成功");
        } catch (Exception e) {
            log.error("创建Kafka主题出错", e);
        }
        return Results.newFailResult(ErrorCode.SYSTEM_ERROR, "创建失败");
    }

    @Override
    public Result<String> sendMessage(String topic, String message) {
        try {
            messageProducer.send(topic, message);
            return Results.newSuccessResult(null, "推送成功");
        } catch (Exception e) {
            log.error("消息推送出错", e);
        }
        return Results.newFailResult(ErrorCode.SYSTEM_ERROR, "推送失败");
    }

    @Override
    public Result<String> deleteTopic(String topic) {
        try {
            Properties prop = new Properties();
            prop.put("bootstrap.servers", "192.168.25.93:9092");
            AdminClient client = AdminClient.create(prop);
            ArrayList<String> topics = new ArrayList<>();
            topics.add(topic);
            DeleteTopicsResult result = client.deleteTopics(topics);
            result.all().get();
            return Results.newSuccessResult(null, "删除成功");
        } catch (Exception e) {
            log.error("删除Kafka主题出错", e);
        }
        return Results.newFailResult(ErrorCode.SYSTEM_ERROR, "删除失败");
    }

    @Override
    public Result<List<String>> topicList() {
        try {
            Properties prop = new Properties();
            prop.put("bootstrap.servers", "192.168.25.93:9092");
            AdminClient admin = AdminClient.create(prop);
            ListTopicsResult topics = admin.listTopics();
            KafkaFuture<Set<String>> future = topics.names();
            List<String> result = new ArrayList<>(future.get());
            return Results.newSuccessResult(result, "查询成功");
        } catch (Exception e) {
            log.error("获取Kafka主题列表出错", e);
        }
        return Results.newFailResult(ErrorCode.SYSTEM_ERROR, "查询失败");
    }
}
